package com.fanjia.rabbitmq_java.topics;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author fanjia
 * @className Consumer
 * @description TODO
 * @date 2021/4/17 15:05
 */
public class Consumer {
   private static Runnable runnable = new Runnable(){

       @Override
       public void run() {

           try {
               //tcp/ip amqp是再tcp/ip再次封装的一种消息协议
               //1.创建连接工程
               ConnectionFactory connectionFactory = new ConnectionFactory();
               connectionFactory.setHost("192.168.130.114");
               connectionFactory.setPort(5672);
               connectionFactory.setUsername("admin");
               connectionFactory.setPassword("admin");
               connectionFactory.setVirtualHost("/");//把我们的消息发在根目录下。

               final String queueName = Thread.currentThread().getName();

               Connection connection = null;
               Channel channel = null;
               //2.创建连接 connection
               connection = connectionFactory.newConnection("消费者");
               //3.通过连接获取通道channel
               try {
                   channel = connection.createChannel();
               } catch (IOException e) {
                   e.printStackTrace();
               }

               //


               //4.消费消息。

               channel.basicConsume(queueName, true, new DeliverCallback() {
                   @Override
                   public void handle(String consumerTag, Delivery message) throws IOException {
                       System.out.println("消息已收到"+new String(message.getBody(),"UTF-8"));
                   }
               },new CancelCallback(){
                   @Override
                   public void handle(String consumerTag) throws IOException {
                       System.out.println("接收失败了");
                   }
               });

               System.out.println("开始接收消息");
               System.in.read();

               //7.关闭连接
               if(connection != null && connection.isOpen()){
                   connection.close();
               }

               //8.关闭通道
               if(channel != null && channel.isOpen()){
                   channel.close();
               }
           }catch (Exception e){
               e.printStackTrace();
           }

       }
   };

    public static void main(String[] args) throws IOException, TimeoutException {
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
        new Thread(runnable,"queue3").start();
    }
}
